/**
 * Copyright (c) 2013-2022 Nikita Koksharov
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson.connection;

import io.netty.buffer.ByteBufUtil;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.EpollDatagramChannel;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.kqueue.KQueueDatagramChannel;
import io.netty.channel.kqueue.KQueueEventLoopGroup;
import io.netty.channel.kqueue.KQueueSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.incubator.channel.uring.IOUringDatagramChannel;
import io.netty.incubator.channel.uring.IOUringEventLoopGroup;
import io.netty.incubator.channel.uring.IOUringSocketChannel;
import io.netty.resolver.AddressResolver;
import io.netty.resolver.AddressResolverGroup;
import io.netty.resolver.DefaultAddressResolverGroup;
import io.netty.resolver.dns.DnsServerAddressStreamProviders;
import io.netty.util.Timer;
import io.netty.util.TimerTask;
import io.netty.util.*;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.*;
import io.netty.util.internal.PlatformDependent;
import org.redisson.ElementsSubscribeService;
import org.redisson.Version;
import org.redisson.api.NatMapper;
import org.redisson.api.RFuture;
import org.redisson.cache.LRUCacheMap;
import org.redisson.client.RedisNodeNotFoundException;
import org.redisson.config.Config;
import org.redisson.config.MasterSlaveServersConfig;
import org.redisson.config.TransportMode;
import org.redisson.misc.CompletableFutureWrapper;
import org.redisson.misc.InfinitySemaphoreLatch;
import org.redisson.misc.RedisURI;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.security.MessageDigest;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

/**
 * 管理redission实例，
 * 它提供创建、获取和删除Redission实例，并对这些实例进行配置和监控
 * 例如集群模式下的主从节点切换、监控实例的状态和性能
 */
public class ServiceManager {

    private final Logger log = LoggerFactory.getLogger(getClass());

    public static final Timeout DUMMY_TIMEOUT = new Timeout() {
        @Override
        public Timer timer() {
            return null;
        }

        @Override
        public TimerTask task() {
            return null;
        }

        @Override
        public boolean isExpired() {
            return false;
        }

        @Override
        public boolean isCancelled() {
            return false;
        }

        @Override
        public boolean cancel() {
            return true;
        }
    };


    /**
     * 监听和管理 Redis 数据库连接事件
     */
    private final ConnectionEventsHub connectionEventsHub = new ConnectionEventsHub();

    /**
     * 使用UUID生成的唯一标识符，用于标识当前实例
     */
    private final String id = UUID.randomUUID().toString();

    /**
     * EventLoopGroup实例，用于处理网络IO事件
     */
    private final EventLoopGroup group;

    /**
     * 地址解析器组，用于解析域名到IP地址
     */
    private final Class<? extends SocketChannel> socketChannelClass;

    private final AddressResolverGroup<InetSocketAddress> resolverGroup;

    private final ExecutorService executor;

    private final Config cfg;

    /**
     * 配置类，设置主从服务器模式下的Redission客户端配置
     */
    private MasterSlaveServersConfig config;

    /**
     * 定义一个私有的HashedWheelTimer变量。
     * HashedWheelTimer是一个用于定时任务的轮询定时器，它可以用来安排未来的任务或者周期性执行任务。
     * 这个变量将被类内部的方法使用，用于管理和调度定时任务。
     */
    private HashedWheelTimer timer;

    /**
     * 这个成员变量是一个IdleConnectionWatcher的实例，
     * 用于监视和管理空闲连接。它在系统中起着重要的作用，
     * 因为它确保了连接池中的空闲连接不会无限期地保持打开状态，
     * 而是根据预定义的策略进行回收和重新利用。
     */
    private IdleConnectionWatcher connectionWatcher;

    private final Promise<Void> shutdownPromise = ImmediateEventExecutor.INSTANCE.newPromise();

    private final InfinitySemaphoreLatch shutdownLatch = new InfinitySemaphoreLatch();

    private final ElementsSubscribeService elementsSubscribeService = new ElementsSubscribeService(this);

    private NatMapper natMapper = NatMapper.direct();

    private static final Map<InetSocketAddress, Set<String>> SCRIPT_SHA_CACHE = new ConcurrentHashMap<>();

    private static final Map<String, String> SHA_CACHE = new LRUCacheMap<>(500, 0, 0);

    // 有参狗子
    /**
     * ServiceManager的构造函数，负责初始化服务管理器。
     * 根据提供的配置初始化网络传输模式，事件循环组，套接字通道类和解析器组。
     * 同时，也初始化执行器线程池，并设置连接监听器。
     * @param cfg 配置对象，包含传输模式、事件循环组、套接字通道配置等初始化所需参数。
     */
    public ServiceManager(Config cfg) {
        Version.logVersion(); // 打印版本信息

        // 根据配置的传输模式选择不同的Netty事件循环组和套接字通道类
        if (cfg.getTransportMode() == TransportMode.EPOLL) {
            // 使用EPOLL传输模式，配置Epoll事件循环组和EpollSocketChannel
            if (cfg.getEventLoopGroup() == null) {
                this.group = new EpollEventLoopGroup(cfg.getNettyThreads(), new DefaultThreadFactory("redisson-netty"));
            } else {
                this.group = cfg.getEventLoopGroup();
            }

            this.socketChannelClass = EpollSocketChannel.class;
            // 根据平台选择合适的解析器组
            if (PlatformDependent.isAndroid()) {
                this.resolverGroup = DefaultAddressResolverGroup.INSTANCE;
            } else {
                this.resolverGroup = cfg.getAddressResolverGroupFactory().create(EpollDatagramChannel.class, DnsServerAddressStreamProviders.platformDefault());
            }
        }

        // 使用KQUEUE传输模式，配置KQueue事件循环组和KQueueSocketChannel
        else if (cfg.getTransportMode() == TransportMode.KQUEUE) {
            if (cfg.getEventLoopGroup() == null) {
                this.group = new KQueueEventLoopGroup(cfg.getNettyThreads(), new DefaultThreadFactory("redisson-netty"));
            } else {
                this.group = cfg.getEventLoopGroup();
            }

            this.socketChannelClass = KQueueSocketChannel.class;
            this.resolverGroup = cfg.getAddressResolverGroupFactory().create(KQueueDatagramChannel.class, DnsServerAddressStreamProviders.platformDefault());
        }

        // 使用IO_URING传输模式，配置IO_URING事件循环组和IOUringSocketChannel
        else if (cfg.getTransportMode() == TransportMode.IO_URING) {

            if (cfg.getEventLoopGroup() == null) {
                this.group = new IOUringEventLoopGroup(cfg.getNettyThreads(), new DefaultThreadFactory("redisson-netty"));
            } else {
                this.group = cfg.getEventLoopGroup();
            }

            this.socketChannelClass = IOUringSocketChannel.class;
            this.resolverGroup = cfg.getAddressResolverGroupFactory().create(IOUringDatagramChannel.class, DnsServerAddressStreamProviders.platformDefault());
        }

        // 默认使用NIO传输模式，配置NIO事件循环组和NioSocketChannel
        else {
            if (cfg.getEventLoopGroup() == null) {
                this.group = new NioEventLoopGroup(cfg.getNettyThreads(), new DefaultThreadFactory("redisson-netty"));
            } else {
                this.group = cfg.getEventLoopGroup();
            }
            this.socketChannelClass = NioSocketChannel.class;
            // 根据平台选择合适的解析器组
            if (PlatformDependent.isAndroid()) {
                this.resolverGroup = DefaultAddressResolverGroup.INSTANCE;
            } else {
                this.resolverGroup = cfg.getAddressResolverGroupFactory().create(NioDatagramChannel.class, DnsServerAddressStreamProviders.platformDefault());
            }
        }

        // 初始化执行器线程池
        if (cfg.getExecutor() == null) {
            int threads = Runtime.getRuntime().availableProcessors() * 2; // 默认使用处理器数量的两倍作为线程数
            if (cfg.getThreads() != 0) {
                threads = cfg.getThreads(); // 如果配置了线程数，则使用配置的线程数
            }
            executor = Executors.newFixedThreadPool(threads, new DefaultThreadFactory("redisson"));
        } else {
            executor = cfg.getExecutor();
        }

        this.cfg = cfg; // 保存配置对象

        // 添加连接监听器
        if (cfg.getConnectionListener() != null) {
            this.connectionEventsHub.addListener(cfg.getConnectionListener());
        }

        // 添加默认的连接事件监听器，用于处理连接断开事件
        this.connectionEventsHub.addListener(new ConnectionListener() {
            @Override
            public void onConnect(InetSocketAddress addr) {
                // 连接建立时的空操作
            }

            @Override
            public void onDisconnect(InetSocketAddress addr) {
                // 连接断开时，从脚本SHA缓存中移除对应的地址
                SCRIPT_SHA_CACHE.remove(addr);
            }
        });
    }


    /**
     * 初始化一个定时器和连接监视器。
     * 此方法不接受参数并且没有返回值。
     * 首先，根据配置的重试间隔和超时时间创建一个整型数组并对其进行排序，
     * 然后根据最小的超时时间来配置定时器的延迟。
     * 定时器采用的是HashedWheelTimer，它基于轮询机制实现。
     * 最后，创建并初始化一个IdleConnectionWatcher用于监视空闲连接。
     */
    public void initTimer() {
        // 根据配置的重试间隔和超时时间，创建并排序一个整型数组
        int[] timeouts = new int[]{config.getRetryInterval(), config.getTimeout()};
        Arrays.sort(timeouts);

        // 计算最小的超时时间，并调整为最合适的值
        int minTimeout = timeouts[0];
        if (minTimeout % 100 != 0) {
            // 如果最小超时时间不是100的倍数，将其除以2
            minTimeout = (minTimeout % 100) / 2;
        } else if (minTimeout == 100) {
            // 如果最小超时时间正好是100，将其设置为50
            minTimeout = 50;
        } else {
            // 如果最小超时时间是100的倍数且不等于100，将其设置为100
            minTimeout = 100;
        }

        // 使用最小超时时间配置HashedWheelTimer
        timer = new HashedWheelTimer(new DefaultThreadFactory("redisson-timer"), minTimeout, TimeUnit.MILLISECONDS, 1024, false);

        // 创建并初始化连接监视器
        connectionWatcher = new IdleConnectionWatcher(group, config);
    }


    /**
     * 创建一个新的超时任务。
     * 此方法通过调用Timer的newTimeout方法来创建一个新的超时任务。如果定时器处于关闭状态，
     * 则会捕获IllegalStateException并根据情况返回一个虚拟的超时对象或重新抛出异常。
     * 这种处理方式确保了在定时器关闭过程中，对newTimeout的调用能够以一种预期的方式处理，
     * 从而避免了程序因为未处理的异常而崩溃。
     */
    public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
        try {
            // 尝试创建一个新的超时任务。
            return timer.newTimeout(task, delay, unit);
        } catch (IllegalStateException e) {
            // 检查定时器是否正在关闭。
            if (isShuttingDown()) {
                // 如果是，返回一个虚拟的超时对象，表示任务将不会被执行。
                return DUMMY_TIMEOUT;
            }
            // 如果不是正在关闭，则重新抛出异常，因为其他原因可能导致了IllegalStateException。
            throw e;
        }
    }


    public boolean isShuttingDown() {
        return shutdownLatch.isClosed();
    }

    public boolean isShutdown() {
        return group.isTerminated();
    }

    public ConnectionEventsHub getConnectionEventsHub() {
        return connectionEventsHub;
    }

    public String getId() {
        return id;
    }

    public EventLoopGroup getGroup() {
        return group;
    }

    public Future<List<InetSocketAddress>> resolveAll(RedisURI uri) {
        AddressResolver<InetSocketAddress> resolver = resolverGroup.getResolver(group.next());
        return resolver.resolveAll(InetSocketAddress.createUnresolved(uri.getHost(), uri.getPort()));
    }

    public AddressResolverGroup<InetSocketAddress> getResolverGroup() {
        return resolverGroup;
    }

    public ExecutorService getExecutor() {
        return executor;
    }

    public Config getCfg() {
        return cfg;
    }

    public HashedWheelTimer getTimer() {
        return timer;
    }

    public IdleConnectionWatcher getConnectionWatcher() {
        return connectionWatcher;
    }

    public Class<? extends SocketChannel> getSocketChannelClass() {
        return socketChannelClass;
    }

    public Promise<Void> getShutdownPromise() {
        return shutdownPromise;
    }

    public InfinitySemaphoreLatch getShutdownLatch() {
        return shutdownLatch;
    }

    public RedisNodeNotFoundException createNodeNotFoundException(NodeSource source) {
        RedisNodeNotFoundException ex;
        if (source.getSlot() != null && source.getAddr() == null && source.getRedisClient() == null) {
            ex = new RedisNodeNotFoundException("Node for slot: " + source.getSlot() + " hasn't been discovered yet. Check cluster slots coverage using CLUSTER NODES command. Increase value of retryAttempts and/or retryInterval settings.");
        } else {
            ex = new RedisNodeNotFoundException("Node: " + source + " hasn't been discovered yet. Increase value of retryAttempts and/or retryInterval settings.");
        }
        return ex;
    }


    public MasterSlaveServersConfig getConfig() {
        return config;
    }

    public void setConfig(MasterSlaveServersConfig config) {
        this.config = config;
    }

    public ElementsSubscribeService getElementsSubscribeService() {
        return elementsSubscribeService;
    }

    public CompletableFuture<RedisURI> resolveIP(RedisURI address) {
        return resolveIP(address.getScheme(), address);
    }

    public CompletableFuture<RedisURI> resolveIP(String scheme, RedisURI address) {
        if (address.isIP()) {
            RedisURI addr = toURI(scheme, address.getHost(), "" + address.getPort());
            return CompletableFuture.completedFuture(addr);
        }

        CompletableFuture<RedisURI> result = new CompletableFuture<>();
        AddressResolver<InetSocketAddress> resolver = resolverGroup.getResolver(group.next());
        InetSocketAddress addr = InetSocketAddress.createUnresolved(address.getHost(), address.getPort());
        Future<InetSocketAddress> future = resolver.resolve(addr);
        future.addListener((FutureListener<InetSocketAddress>) f -> {
            if (!f.isSuccess()) {
                log.error("Unable to resolve {}", address, f.cause());
                result.completeExceptionally(f.cause());
                return;
            }

            InetSocketAddress s = f.getNow();
            RedisURI uri = toURI(scheme, s.getAddress().getHostAddress(), "" + address.getPort());
            result.complete(uri);
        });
        return result;
    }

    public RedisURI toURI(String scheme, String host, String port) {
        // convert IPv6 address to unified compressed format
        if (NetUtil.isValidIpV6Address(host)) {
            byte[] addr = NetUtil.createByteArrayFromIpAddressString(host);
            try {
                InetAddress ia = InetAddress.getByAddress(host, addr);
                host = ia.getHostAddress();
            } catch (UnknownHostException e) {
                throw new RuntimeException(e);
            }
        }
        RedisURI uri = new RedisURI(scheme + "://" + host + ":" + port);
        return natMapper.map(uri);
    }

    public void setNatMapper(NatMapper natMapper) {
        this.natMapper = natMapper;
    }

    public boolean isCached(InetSocketAddress addr, String script) {
        Set<String> values = SCRIPT_SHA_CACHE.computeIfAbsent(addr, k -> Collections.newSetFromMap(new ConcurrentHashMap<>()));
        String sha = calcSHA(script);
        return values.contains(sha);
    }

    public void cacheScripts(InetSocketAddress addr, Set<String> scripts) {
        Set<String> values = SCRIPT_SHA_CACHE.computeIfAbsent(addr, k -> Collections.newSetFromMap(new ConcurrentHashMap<>()));
        for (String script : scripts) {
            values.add(calcSHA(script));
        }
    }

    public String calcSHA(String script) {
        return SHA_CACHE.computeIfAbsent(script, k -> {
            try {
                MessageDigest mdigest = MessageDigest.getInstance("SHA-1");
                byte[] s = mdigest.digest(script.getBytes());
                return ByteBufUtil.hexDump(s);
            } catch (Exception e) {
                throw new IllegalStateException(e);
            }
        });
    }

    public <T> RFuture<T> execute(Supplier<CompletionStage<T>> supplier) {
        CompletableFuture<T> result = new CompletableFuture<>();
        int retryAttempts = config.getRetryAttempts();
        AtomicInteger attempts = new AtomicInteger(retryAttempts);
        execute(attempts, result, supplier);
        return new CompletableFutureWrapper<>(result);
    }

    private <T> void execute(AtomicInteger attempts, CompletableFuture<T> result, Supplier<CompletionStage<T>> supplier) {
        CompletionStage<T> future = supplier.get();
        future.whenComplete((r, e) -> {
            if (e != null) {
                if (e.getCause().getMessage().equals("None of slaves were synced")) {
                    if (attempts.decrementAndGet() < 0) {
                        result.completeExceptionally(e);
                        return;
                    }

                    newTimeout(t -> execute(attempts, result, supplier),
                            config.getRetryInterval(), TimeUnit.MILLISECONDS);
                    return;
                }

                result.completeExceptionally(e);
                return;
            }

            result.complete(r);
        });
    }

    public <V> void transfer(CompletionStage<V> source, CompletableFuture<V> dest) {
        source.whenComplete((res, e) -> {
            if (e != null) {
                dest.completeExceptionally(e);
                return;
            }

            dest.complete(res);
        });
    }

    public String generateId() {
        return ByteBufUtil.hexDump(generateIdArray());
    }

    public byte[] generateIdArray() {
        return generateIdArray(16);
    }

    public byte[] generateIdArray(int size) {
        byte[] id = new byte[size];
        ThreadLocalRandom.current().nextBytes(id);
        return id;
    }

    private final AtomicBoolean liveObjectLatch = new AtomicBoolean();

    public AtomicBoolean getLiveObjectLatch() {
        return liveObjectLatch;
    }
}
